package com.xzx.flink.streamapi.window;

import com.xzx.flink.bean.ClickEvent;
import com.xzx.flink.streamapi.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.client.program.StreamContextEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

/**
 * 计算每个 url 在 10 秒滚动窗口的 pv 指标，然后设置触发器，每隔 1 秒钟触发一次窗口的计算。
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/5/3 14:41
 */
public class Window_05_Trigger {
    private static final Logger log = LoggerFactory.getLogger(Window_05_Trigger.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamContextEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                return element.getTimestamp();
                            }
                        }))
                .keyBy(ClickEvent::getUrl)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .trigger(new Trigger<ClickEvent, TimeWindow>() {
                    @Override
                    public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        // 标记是不是这个窗口里面的第一条消息
                        ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-event", Types.BOOLEAN));
                        if (isFirstEvent.value() == null) {
                            // 状态里面不存在，说明是这个窗口的第一条消息，给这个窗口的每一秒注册一个定时器
                            for (long i = window.getStart(); i < window.getEnd(); i = i + 1000L) {
                                ctx.registerEventTimeTimer(i);
                            }
                            isFirstEvent.update(true);
                        }
                        return TriggerResult.CONTINUE;
                    }

                    // 当前的这列例子里面这个方法不会被执行，因为没有注册过处理时间定时器
                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        log.info("==============onProcessingTime======{}", new Timestamp(time));
                        return TriggerResult.CONTINUE;
                    }

                    // 当自己注册的事件时间定时器到点时触发
                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        log.info("==============onEventTime======{}", new Timestamp(time));
                        return TriggerResult.FIRE;
                    }

                    // 窗口关闭的时候清除状态
                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
                        ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(new ValueStateDescriptor<Boolean>("first-event", Types.BOOLEAN));
                        isFirstEvent.clear();
                    }
                })
                .aggregate(new AggregateFunction<ClickEvent, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(ClickEvent value, Long accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return a + b;
                    }
                }, new ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>() {
                    @Override
                    public void process(String key, ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {
                        UrlViewCount urlViewCount = new UrlViewCount();
                        urlViewCount.setWindowStart(context.window().getStart());
                        urlViewCount.setWindowEnd(context.window().getEnd());
                        urlViewCount.setUrl(key);
                        urlViewCount.setCount(elements.iterator().next());
                        out.collect(urlViewCount);
                    }
                })
//                .process(new ProcessWindowFunction<ClickEvent, UrlViewCount, String, TimeWindow>() {
//                    @Override
//                    public void process(String key, ProcessWindowFunction<ClickEvent, UrlViewCount, String, TimeWindow>.Context context, Iterable<ClickEvent> elements, Collector<UrlViewCount> out) throws Exception {
//                        UrlViewCount urlViewCount = new UrlViewCount();
//                        urlViewCount.setWindowStart(context.window().getStart());
//                        urlViewCount.setWindowEnd(context.window().getEnd());
//                        urlViewCount.setUrl(key);
//                        // 获取迭代器中的元素个数
//                        urlViewCount.setCount(elements.spliterator().getExactSizeIfKnown());
//                        out.collect(urlViewCount);
//                    }
//                })
                .print();


        env.execute(Window_05_Trigger.class.getSimpleName());
    }
}
